package com.lagou.edu.utils;

import com.alibaba.fastjson.JSON;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.nio.charset.StandardCharsets;

/**
 * @author 赵虎
 * @date 2020/11/29 6:58 下午
 */
public class ZookeeperFactory {

    private static ZookeeperFactory zookeeperFactory;

    private static String zkConnect = "10.10.10.205:2181";

    private static final String NAME_SPACE = "lg-database";

    private static final String addressSubMark = ":";

    private static final String urlSubMark = "/";

    public static final String zkNodeName = "/db";

    private CuratorFramework client;

    public static ZookeeperFactory getInstance(String ip, int port) {
        if (zookeeperFactory == null) {
            zookeeperFactory = new ZookeeperFactory(ip, port);
        }
        return zookeeperFactory;
    }

    public static ZookeeperFactory getInstance() {
        if (zookeeperFactory == null) {
            zookeeperFactory = new ZookeeperFactory();
        }
        return zookeeperFactory;
    }

    public ZookeeperFactory() {
        initClient();
    }

    public ZookeeperFactory(String ip, int port) {
        zkConnect = StringUtils.join(ip, addressSubMark, port);
        initClient();
    }


    /**
     * 初始化Zookeeper客户端
     */
    private void initClient() {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        client = CuratorFrameworkFactory.builder()
                .connectString(zkConnect)
                .sessionTimeoutMs(50000)
                .connectionTimeoutMs(30000)
                .retryPolicy(retryPolicy)
                .namespace(NAME_SPACE)
                .build();
        client.start();
    }

    /**
     * 获得服务节点的子节点监听对象
     */
    public PathChildrenCache getPathChildrenCache(String path) throws Exception {
        PathChildrenCache cache = new PathChildrenCache(client, StringUtils.join(urlSubMark, path), true);
        cache.start();
        return cache;
    }

    /**
     * 获得服务节点的子节点监听对象
     */
    public void createPath(String path,Object object) throws Exception {
        if(!checkExists(path)){
            client.create().creatingParentsIfNeeded().forPath(path,JSON.toJSONString(object).getBytes(StandardCharsets.UTF_8));
        }
    }


    /**
     * 获取路径值
     */
    public Boolean checkExists(String path) throws Exception {
        return client.checkExists().forPath(path) != null;
    }

    /**
     * 获得服务节点的子节点监听对象
     */
    public <T> T getPathValue(String path, Class<T> clazz) throws Exception {
        if(checkExists(path)){
            byte[] bytes = client.getData().forPath(path);
            String valueString = new String(bytes);
            return JSON.parseObject(valueString, clazz);
        }
        return null;
    }

    /**
     * 获得服务节点的子节点监听对象
     */
    public void setPathValue(String path, Object object) throws Exception {
        if(!checkExists(path)){
            createPath(path,null);
        }
        client.setData().forPath(path,JSON.toJSONString(object).getBytes(StandardCharsets.UTF_8));
    }

    /**
     * 获得节点监听对象
     */
    public NodeCache getNodeCache(String path) throws Exception {
        NodeCache nodeCache = new NodeCache(client, path, false);
        nodeCache.start();
        return nodeCache;
    }
}
